package com.shopee.am.tool;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;


/**
 * @author:huanqing.cheng
 * @time:2022/1/25 5:37 下午
 * @description: HBase工具类
 */
public class HBaseTool {

    private static final Logger logger = LoggerFactory.getLogger(HBaseTool.class);
    private static Connection connection;
    private static Configuration configuration;
    private static BufferedMutator mutator;
    private static final String tableName = "TEST:monitor_am__task_info";

    /**
     * 初始化HBase-ZK配置
     */
    private static Configuration initHbaseConfig() {
        try {
            configuration = HBaseConfiguration.create();
            configuration.set("hbase.zookeeper.quorum", "hbase00C-test.data-infra.shopee.io,hbase00D-test.data-infra.shopee.io,hbase00E-test.data-infra.shopee.io");
            configuration.set("zookeeper.znode.parent", "/dev-hbase");
        } catch (Exception e) {
            logger.error("初始化HBaseConfig失败: ", e);
        }
        return configuration;
    }

    /**
     * 初始化HBase连接
     */
    public static void initHBaseConn() {
        try {
            connection = ConnectionFactory.createConnection(initHbaseConfig());
            mutator = connection.getBufferedMutator(TableName.valueOf(tableName));
            System.out.println("初始化HBase环境成功: " + connection);
        } catch (Exception e) {
            logger.error("初始化HBase连接失败: ", e);
        }
    }

    /**
     * 异步批量写入HBase
     */
    public static void putData(ArrayList<Put> puts) {
        if (connection == null) initHBaseConn();

        try {
            mutator.mutate(puts);
        } catch (Exception e) {
            logger.error("批量写入HBase失败: ", e);
        }
    }

    /**
     * 创建hbase-put
     */
    public static Put createPut(String rowKey, String columnFamily, String columnName, String value) {
        return new Put(Bytes.toBytes(rowKey)).addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName), Bytes.toBytes(value));
    }

    /**
     * HBase资源关闭
     */
    public static void close() {
        try {
            if (mutator != null) mutator.close();
            if (connection != null) connection.close();
            System.out.println("HBase资源关闭成功");
        } catch (Exception e) {
            logger.error("HBase资源关闭失败: ", e);
        }
    }

}
